package com.gis507.mapreduce.analysis_index;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.AISBean.AnalysisIndex;
import com.gis507.mapreduce.Utils.UtilsClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;

@Slf4j
public class AISInterpolationReducer extends Reducer<Text, AnalysisIndex,AnalysisIndex, NullWritable> {

//    //设定的最大时间
//    private static final double STANDTIME = 30;
//    private static final double MinTime = 30;
//    //设定的最大距离
//    private static final double STANDSTANCE = 1000;
//    private static int sum = 0;
    private double maxStance = 500;

    @Override
    protected void reduce(Text key, Iterable<AnalysisIndex> values, Context context) throws IOException, InterruptedException {

        ArrayList<AnalysisIndex> beanList = new ArrayList<>(10000);

        for (AnalysisIndex value : values) {
            beanList.add(UtilsClass.getNewAnalysisIndex(value));
        }
        if (beanList.size() > 5500){
            Collections.sort(beanList);

            for (int i = 1; i < beanList.size(); i++) {
                Double distance = UtilsClass.getDecimal(6,UtilsClass.getDisStance(beanList.get(i), beanList.get(i-1)));

                Double diffSpeed = UtilsClass.getDecimal(2,beanList.get(i).getSpeed() - beanList.get(i-1).getSpeed());
                Double diffHeading = UtilsClass.getDecimal(2,beanList.get(i).getHeading() - beanList.get(i-1).getHeading());
                Long diffTime = beanList.get(i).getUnixTime() - beanList.get(i-1).getUnixTime();
                Double diffCourse = UtilsClass.getDecimal(2,beanList.get(i).getCourse() - beanList.get(i-1).getCourse());

                AnalysisIndex ais = UtilsClass.getNewAnalysisIndex(beanList.get(i));
                ais.setDistance(distance);
                ais.setDiffTime(diffTime);
                ais.setDiffSpeed(diffSpeed);
                ais.setDiffHeading(diffHeading);
                ais.setDiffCourse(diffCourse);
                context.write(ais,NullWritable.get());
                //1511.5846662461386,-1610,-5.9,0.0,-4.100000000000023
        }



        }

    }
}
